package demo.mq.rocket;

import demo.java.lang.ThreadDemo;
import demo.mq.rocket.client.ConcurrentlyMessageListenerDemo;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageQueueListener;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * 消费过程要做到幂等。在编程中一个幂等操作的特点是其任意多次执行所产生的影响均与一次执行的影响相同。
 * 幂等函数，或幂等方法，是指可以使用相同参数重复执行，并能获得相同结果的函数。这些函数不会影响系统状态，也不用担心重复执行会对系统造成改变。
 * <p>
 * RocketMQ目前无法避免消息重复，所以如果业务对消费重复非常敏感，务必要在业务层面去重，有以下几种去重方式：
 *
 * <li>将消息的唯一键，可以是msgId，也可以是消息内容中的唯一标识字段，例如订单Id等，消费之前判断是否在Db或Tair(全局KV存储)中存在，如果不存在则插入，并消费，否则跳过。
 * （实际过程要考虑原子性问题，判断是否存在可以尝试插入，如果报主键冲突，则插入失败，直接跳过）。
 * msgId一定是全局唯一标识符，但是可能会存在同样的消息有两个不同msgId的情况（有多种原因），这种情况可能会使业务上重复消费，建议最好使用消息内容中的唯一标识字段去重。
 * <li>使用业务层面的状态机去重。
 *
 * <h2>提高消费并行度</h2><br>
 * 绝大部分消息消费行为属于IO密集型，即可能是操作数据库，或者调用RPC，这类消费行为的消费速度在于后端数据库或者外系统的吞吐量，
 * 通过增加消费并行度，可以提高总的消费吞吐量，但是并行度增加到一定程度，反而会下降。
 *
 * <h3>修改消费并行度方法如下所示：</h3>
 * <li>同一个ConsumerGroup下，通过增加Consumer实例数量来提高并行度，超过订阅队列数的Consumer实例无效。
 * 可以通过加机器，或者在已有机器启动多个进程的方式。
 * <li>提高单个Consumer的消费并行线程，通过修改以下参数.consumer.setConsumeThreadMin();consumer.setConsumeThreadMax()
 *
 * <h2>消息批量消费</h2><br>
 * 某些业务流程如果支持批量方式消费，则可以很大程度上提高消费吞吐量，例如订单扣款类应用，一次处理一个订单耗时1秒钟，一次处理10个订单可能也只耗时2秒钟，这样即可大幅度提高消费的吞吐量。
 * 通过设置consumer的consumeMessageBatchMaxSize这个参数，默认是1，即一次只消费一条消息，例如设置为N，那么每次消费的消息数小于等于N。
 *
 * <h2>跳过非重要消息</h2><br>
 * 发生消息堆积时，如果消费速度一直追不上发送速度，可以选择丢弃不重要的消息
 *
 * <h2>优化每条消息消费过程</h2>
 * <h1>2.3 消费打印日志</h1>如果消息量较少，建议在消费入口方法打印消息，消费耗时等，方便后续排查问题。
 * <h1>2.4 其他消费建议</h1>
 * <li>1 关于消费者和订阅： ​第一件需要注意的事情是，不同的消费者组可以独立的消费一些
 * topic，并且每个消费者组都有自己的消费偏移量，请确保同一组内的每个消费者订阅信息保持一致。
 *
 * <li>2 关于有序消息： 消费者将锁定每个消息队列，以确保他们被逐个消费，虽然这将会导致性能下降，但是当你关心消息顺序的时候会很有用。
 * 我们不建议抛出异常，你可以返回 ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT 作为替代。
 *
 * <li>3 关于并发消费： 顾名思义，消费者将并发消费这些消息，建议你使用它来获得良好性能，我们不建议抛出异常，你可以返回
 * ConsumeConcurrentlyStatus.RECONSUME_LATER 作为替代。
 *
 * <li>4 关于消费状态Consume Status： 对于并发的消费监听器，你可以返回 RECONSUME_LATER
 * 来通知消费者现在不能消费这条消息，并且希望可以稍后重新消费它。 然后，你可以继续消费其他消息。对于有序的消息监听器，因为你关心它的顺序，所以不能跳过消息，
 * 但是你可以返回SUSPEND_CURRENT_QUEUE_A_MOMENT 告诉消费者等待片刻。
 *
 * <li>5 关于Blocking： 不建议阻塞监听器，因为它会阻塞线程池，并最终可能会终止消费进程
 *
 * <li>6 关于线程数设置： 消费者使用 ThreadPoolExecutor 在内部对消息进行消费，所以你可以通过设置
 * setConsumeThreadMin 或 setConsumeThreadMax 来改变它。
 *
 * <li>7 关于消费位点： 当建立一个新的消费者组时，需要决定是否需要消费已经存在于 Broker中的历史消息。
 * CONSUME_FROM_LAST_OFFSET 将会忽略历史消息，并消费之后生成的任何消息。
 * CONSUME_FROM_FIRST_OFFSET将会消费每个存在于 Broker 中的信息。 你也可以使用 CONSUME_FROM_TIMESTAMP
 * 来消费在指定时间戳后产生的消息。
 */
public class ConsumerDemo {

    private static Logger logger = LoggerFactory.getLogger(ConsumerDemo.class);

    public static final String ProducerGroup = "group_xdcs_c";

    @Test
    public void consumer() {
        String topic2 = ProducerDemo.TOPIC_XDCS_2;
        topic2 = "liyuqin_test";
        String topic = "topic_round_lottery_result_jungle";
        try {
            // 实例化消费者
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(ProducerGroup);

            // 设置NameServer的地址
            consumer.setNamesrvAddr(ProducerDemo.NameSvrAddr);

            // 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
            consumer.subscribe(topic, "*");
            consumer.subscribe(topic2, "*");
            // 注册回调实现类来处理从broker拉取回来的消息
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                                ConsumeConcurrentlyContext context) {
                    System.out.printf("[%s] Receive New Messages，msgs.size=%s %n",
                            Thread.currentThread().getName(), msgs.size());

                    for (MessageExt messageExt : msgs) {
                        String tags = messageExt.getTags();
                        System.out.println(messageExt);
                        String msg = new String(messageExt.getBody(), StandardCharsets.UTF_8);
                        System.out.println(msg);
                        System.out.println();
                        if ("XXXXXXX".equals(tags)) {
                            return ConsumeConcurrentlyStatus.RECONSUME_LATER;

                        } else {
                            // 标记该消息已经被成功消费
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        }
                    }
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            });
            // 启动消费者实例
            consumer.start();
            System.out.printf("Consumer Started.%n");
            ThreadDemo.safeSleep(600 * 1000L);
        } catch (Exception e) {
            logger.error("", e);
        }

    }


    /**
     * 通过拉去的方式来消费消息
     *
     * @throws MQClientException
     */
    @Test
    public void testPullConsumer() {
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer();
        consumer.setNamesrvAddr("100.66.154.81:9876");
        consumer.setConsumerGroup("broker");
        try {
            consumer.start();
            Set<MessageQueue> messageQueues = consumer.fetchSubscribeMessageQueues("PushTopic");

            for (MessageQueue messageQueue : messageQueues) {

                System.out.println(messageQueue.getTopic());
            }

            // 消息队列的监听
            consumer.registerMessageQueueListener("", new MessageQueueListener() {

                @Override
                // 消息队列有改变，就会触发
                public void messageQueueChanged(String topic, Set<MessageQueue> mqAll, Set<MessageQueue> mqDivided) {
                    // TODO Auto-generated method stub

                }
            });

        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    /**
     * 从名字上已经可以看出其消息获取方式为broker往消费端推送数据，其内部实现了流控，消费位置上报等等。
     * 通过注册监听的方式来消费信息。 当前例子是PushConsumer用法，使用方式给用户感觉是消息从RocketMQ服务器推到了应用客户端。
     * 但是实际PushConsumer内部是使用长轮询Pull方式从Broker拉消息，然后再回调用户Listener方法<br>
     *
     * @throws MQClientException
     */
    @Test
    public void testPushConsumer() throws MQClientException {
        /**
         * 一个应用创建一个Consumer，由应用来维护此对象，可以设置为全局对象或者单例。
         */
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
        /**
         * 消费者组名，必须设置。 需要注意的是，多个消费者如果具有同样的组名，那么这些消费者必须只消费同一个topic。 一个consumerGroup只对应一个topic
         */
        consumer.setConsumerGroup("GROUP_JAVA_DEMO");
        /**
         * 消费的方式，分为两种：
         * <li>BROADCASTING 广播模式，即所有的消费者可以消费同样的消息
         * <li>CLUSTERING 集群模式，即所有的消费者平均来消费一组消息
         */
        consumer.setMessageModel(MessageModel.CLUSTERING);
        /**
         * ConsumeFromWhere 消费者从那个位置消费，分别为：
         * <li>CONSUME_FROM_LAST_OFFSET：第一次启动从队列最后位置消费，后续再启动接着上次消费的进度开始消费
         * <li>CONSUME_FROM_FIRST_OFFSET：第一次启动从队列初始位置消费，后续再启动接着上次消费的进度开始消费
         * <li>CONSUME_FROM_TIMESTAMP：第一次启动从指定时间点位置消费，后续再启动接着上次消费的进度开始消费
         * <p>
         * 以上所说的第一次启动是指从来没有消费过的消费者，如果该消费者消费过，那么会在broker端记录该消费者的消费位置，如果该消费者挂了再启动，那么自动从上次消费的进度开始
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        /**
         * AllocateMessageQueueStrategy 消息分配策略，用于集群模式下，消息平均分配给所有客户端。默认实现为AllocateMessageQueueAveragely
         */
        consumer.setAllocateMessageQueueStrategy(new AllocateMessageQueueAveragely());
        /**
         * subscription // topic对应的订阅tag
         */
        Map<String, String> subscription = new HashMap<>();
        subscription.put("TOPIC_JAVA_DEMO", "*");
        consumer.setSubscription(subscription);
        /**
         * offset存储实现，分为本地存储或远程存储。集群消费默认为RemoteBrokerOffsetStore
         */
        consumer.setOffsetStore(null);

        consumer.setNamesrvAddr("172.16.30.162:9876");

        /**
         * 每1分钟调整一次线程池，这也是针对消费者来说的，具体为如果消息堆积超过10W条，则调大线程池，最多64个线程；如果消息堆积少于8W条，则调小线程池，最少20的线程。
         */
        consumer.setConsumeThreadMin(20);
        consumer.setConsumeThreadMax(64);
        consumer.setAdjustThreadPoolNumsThreshold(10_0000);
        /**
         * 单队列并行消费最大跨度，用于流控
         */
        consumer.setConsumeConcurrentlyMaxSpan(2000);
        /**
         * 一个queue最大消费的消息个数，用于流控
         */
        consumer.setPullThresholdForQueue(1000);
        /**
         * 消息拉取时间间隔，默认为0，即拉完一次立马拉第二次，单位毫秒
         */
        consumer.setPullInterval(0);
        /**
         * 并发消费时，一次消费消息的数量，默认为1，假如修改为50，此时若有100条消息，那么会创建两个线程，每个线程分配50条消息。
         */
        consumer.setConsumeMessageBatchMaxSize(1);
        /**
         * 消息拉取一次的数量
         */
        consumer.setPullBatchSize(32);

        consumer.setPostSubscriptionWhenPull(false);
        consumer.setUnitMode(false);

        /** 订阅指定topic下tags分别等于TagA或TagB */
        consumer.subscribe("broker-a", "TagB || TagA");
        /**
         * 订阅指定topic下所有消息。 注意：一个consumer对象可以订阅多个topic
         */
        consumer.subscribe("TopicTest2", "*");
        /**
         * 注册消息回调，如果需要顺序消费，需要注册MessageListenerOrderly的实现 。客户端消费消息的实现类
         */
        consumer.registerMessageListener(new ConcurrentlyMessageListenerDemo());

        /**
         * 启动消息消费，初始化一次即可
         */
        consumer.start();

        System.out.println("Consumer Started.");

        /**
         * Shut down once the consumer instance is not longer in use.
         */
        consumer.shutdown();
    }
}
